const path = require("path");
const { Worker } = require("worker_threads");
const fs = require("fs");
class MainWorker {
  workersMap = {};
  works = [];
  stop = false;
  batchNum = 1000;
  logDirPath;
  tbName;
  loadedNum = 0;
  workerNum = 4;
  errorCount = 0;
  total = 0;
  constructor({ workerNum = 4, batchNum = 1000, logDirPath }) {
    this.batchNum = batchNum;
    this.logDirPath = logDirPath;
    this.workerNum = workerNum;
    new Array(workerNum).fill(0).map(() => {
      const worker = new Worker(path.join(__dirname, "./worker.js"));
      this.workersMap[worker.threadId] = { worker, status: 0 };
      worker.on("message", ({ msgType, payload }) => {
        if (msgType === "statusChange") {
          this.onWorkerStatusChange(worker, payload);
        } else if (msgType === "iambusy") {
          this.onWorkerBusy(worker, payload);
        } else if (msgType === "load") {
          this.loadedNum += 1;
        } else if (msgType === "errorCount") {
          this.errorCount += payload;
        }
      });
    });
  }
  __checkLoaded() {
    return new Promise((resolve) => {
      if (this.loadedNum !== this.workerNum) {
        setTimeout(() => {
          this.__checkLoaded().then(resolve);
        }, 10);
      } else {
        resolve();
      }
    });
  }
  initSequlize({ seqConfig, modelDirPath, csvDirPath, tbNames, csvExtension }) {
    return new Promise((resolve) => {
      Object.values(this.workersMap).forEach(({ worker }) => {
        worker.postMessage({
          msgType: "init",
          payload: {
            seqConfig,
            modelDirPath,
            csvDirPath,
            threadId: worker.threadId,
            tbNames,
            csvExtension,
          },
        });
        this.__checkLoaded().then(resolve);
      });
    });
  }

  // 当发送工作给worker，worker 却还有其他工作时，会退回分配的工作
  onWorkerBusy(worker, datas) {
    this.total -= datas.length;
    const workerState = this.workersMap[worker.threadId];
    if (!workerState) this.workersMap[worker.threadId] = { worker, status: 1 };
    this.works = [...datas, ...this.works];
  }
  // 当worker的工作状态变化时
  onWorkerStatusChange(worker, status) {
    // value 1 正在工作 0 空闲状态
    const workerState = this.workersMap[worker.threadId];
    if (!workerState) this.workersMap[worker.threadId] = { worker, status };
    else workerState.status = status;
    if (!status && this.works.length > 0) {
      this.__driveToWork(worker);
    } else if (!status && this.stop) {
      worker.terminate();
    }
  }

  // 分配工作
  reciveWorks(datas = []) {
    return new Promise((resolve) => {
      const workerThd = Object.values(this.workersMap).find(
        (wt) => wt.status === 0
      );
      if (workerThd && this.works.length < this.batchNum) {
        if (!Array.isArray(datas)) this.works.push(datas);
        else if (datas.length > 0) this.works = [...this.works, ...datas];
        const worker = workerThd.worker;
        workerThd.status = 1;
        this.__driveToWork(worker);
        resolve();
      } else {
        setTimeout(() => {
          this.reciveWorks(datas).then(resolve);
        }, 10);
      }
    });
  }
  // 给工作线程分配工作
  __driveToWork(worker) {
    const list = this.works.splice(0, this.batchNum);
    this.total += list.length;
    worker.postMessage({
      msgType: "gotoWork",
      payload: {
        data: list,
        logDirPath: this.logDirPath,
        tbName: this.tbName,
      },
    });
  }

  async stopWork(force) {
    await this.waitFinished();
    this.__recoderErrorCount();
    process.nextTick(() => {
      this.stop = true;
      Object.values(this.workersMap).forEach(({ worker, status }) => {
        if (force) worker.terminate();
        else if (!status) worker.terminate();
      });
    });
  }

  __checkIsFinished() {
    return (
      this.works.length === 0 &&
      Object.values(this.workersMap).some(({ status }) => status)
    );
  }

  waitFinished() {
    return new Promise((resolve) => {
      if (this.__checkIsFinished()) {
        setTimeout(() => {
          this.waitFinished().then(resolve);
        }, 100);
      } else {
        resolve();
      }
    });
  }
  registerTable(tbName) {
    return new Promise((resolve) => {
      this.__checkLoaded().then(() => {
        this.waitFinished().then(() => {
          if (this.tbName && this.tbName !== tbName) {
            this.__recoderErrorCount();
          }
          this.tbName = tbName;
          resolve();
        });
      });
    });
  }

  __recoderErrorCount() {
    writeLog(this.tbName, this.errorCount, this.total, this.logDirPath);
    this.errorCount = 0;
    this.total = 0;
  }
}

function writeLog(tbName, errorCount, total, logDirPath) {
  const dirPath = path.join(process.cwd(), logDirPath);
  if (!fs.existsSync(dirPath)) {
    throw new Error(`请先创建错误日志目录 ${dirPath}`);
  }
  const filePath = path.join(dirPath, "errorCount.log");
  fs.appendFileSync(
    filePath,
    `\n${tbName}总数：${total} 错误条数：${errorCount}\n`
  );
}

module.exports = MainWorker;
